package com.patsnap.data.npd.dw.core.pipeline;

import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.patsnap.data.npd.dw.repository.DBOperator;
import com.patsnap.one.etl.cdc.CdcRecord;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public abstract class AbstractPipeline<T extends PipelineContext> implements Serializable {
    protected Map<String, String> config = new HashMap<>();

    public abstract String pipelineName();

    public abstract T initContext(CdcRecord cdcRecord);

    public void initConfig(Map<String, String> map) {
        config.putAll(map);
    }

    public abstract void extract(DBOperator client, T context) throws PipelineException;

    public abstract void transform(DBOperator client, T context) throws PipelineException;

    public T transformWindow(DBOperator client, String key, Iterable<T> contextIterable) throws PipelineException {
        try {
            return Iterables.getLast(contextIterable);
        } catch (Exception e) {
            throw new PipelineException(e);
        }
    }

    public List<T> flatmap(DBOperator client, T context) throws PipelineException {
        return Lists.newArrayList(context);
    }

    public abstract void load(DBOperator client, T context) throws PipelineException;
}
